package main.java.com.itlin.datastructure.blockingQueue;

import java.util.Arrays;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 单锁实现
 * @param <E> - 元素类型
 */
public class BlockingQueue2<E> implements BlockingQueue<E> {

    private final E[] array;
    private int head;
    private int tail;
    private AtomicInteger size = new AtomicInteger();

    private ReentrantLock tailLock = new ReentrantLock();
    private Condition tailWaits = tailLock.newCondition();
    private ReentrantLock headLock = new ReentrantLock();
    private Condition headWaits = headLock.newCondition();


    @SuppressWarnings("all")
    public BlockingQueue2(int capacity) {
        array = (E[]) new Object[capacity];
    }


    private boolean isEmpty() {
        return size.get() == 0;
    }

    private boolean isFull() {
        return size.get() == array.length;
    }
    @Override
    public void offer(E e) throws InterruptedException { // poll在 等待队列非空
        //加锁
        tailLock.lockInterruptibly();
        int c;// 添加前的元素个数
        try{
            //1.队列满则等待
            while (isFull()){
                tailWaits.await(); //offer_1  offer_2
            }
            //2.队列不满则 入队
            array[tail] = e;
            if(++tail == array.length){
                tail = 0;
            }
            //3. 修改size
            /*
                size = 5
             */
            c = size.getAndIncrement();
            if (c + 1 < array.length){
                //offer_1
                tailWaits.signal();
            }
            /*
                1. 读取成员变量size的值
                2.自增
                3. 结果写回到成员变量
             */
            //
        }finally {
            tailLock.unlock();
        }
        //4. 唤醒等待非空的 poll线程
        //                           0->1       1->2     2->3
        if (c == 0) {
            headLock.lock();            //offer_1 offer_2 offer_3
            try {
                headWaits.signal(); // poll在等待 非空的线程开启
            } finally {
                headLock.unlock();
            }
        }
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException { // 毫秒  5s
        //加锁
        headLock.lockInterruptibly();

        try{
            long t = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isFull()){
                if (t <= 0 ){
                    return false;
                }
                t = tailWaits.awaitNanos(t);//最多等待多少ns纳秒  1s  4s  返回值 返回等待的剩余时间
            }
            array[tail] = e;
            if(++tail == array.length){
                tail = 0;
            }
            size.getAndIncrement();
        }finally {
            headLock.unlock();
        }
        try{
            headWaits.signal(); // poll在等待 非空的线程开启
            return true;
        }finally {
            headLock.unlock();
        }
    }

    @Override
    public E poll() throws InterruptedException {// offer 在 等待队列 非满
        E e;
        int c;// 取走元素前的 元素个数

        //加锁
        headLock.lockInterruptibly();
        try {
            //如果队列为空，则等待
            while (isEmpty()){
                headWaits.await(); //poll_1,poll_2
            }
            //如果队列不空，则出队
            e = array[head];
            array[head] = null;
            if (++head == array.length){
                head = 0;
            }
            // 修改size
            c = size.getAndDecrement();
            /*
                1. 读取成员变量size的值
                2.自增
                3. 结果写回到成员变量
             */
            //poll_1  poll_2  poll_3
            if (c > 1){// 3 -> 2    2 -> 1   1 -> 0
                headWaits.signal();
            }

        }finally {
            headLock.unlock();
        }
        //4. 队列从 满 -> 不满时， 由 poll 唤醒等待不满的offer线程
        if (c == array.length) {
            tailLock.lock();
            try {
                tailWaits.signal(); // offer在等待 非满的线程开启
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }





    @Override
    public String toString() {
        return "BlockingQueue1{" +
                "array=" + Arrays.toString(array) +
                '}';
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue2<String> queue = new BlockingQueue2<>(3);
        queue.offer("任务1");

        new Thread(() -> {
            try {
                queue.offer("任务2");
            }catch (InterruptedException e){
                throw new RuntimeException(e);
            }
        },"offer").start();

        new Thread(() -> {
            try {
                System.out.println(queue.poll());
            }catch (InterruptedException e){
                throw new RuntimeException(e);
            }
        },"poll").start();

    }
}
